package com.sky.consumer;

import com.rabbitmq.client.Channel;
import io.lettuce.core.RedisCommandTimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Set;

@Slf4j
@Component
public class CacheDelConsumer {
    private static final String CACHE_TO_DEL_QUEUE="cache_to_del_queue";
    private static final String CACHE_TO_DEL_EXCHANGE="cache_del_exchange";

    private static final int MAX_RETRY_COUNT=2;

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = CACHE_TO_DEL_QUEUE,ackMode = "MANUAL")
    public void delCacheInRedis(String pattern, Channel channel, Message message) throws IOException {
        try {
            log.info("del redis cache.");
            Set keys = redisTemplate.keys(pattern);
            redisTemplate.delete(keys);
            //throw new RedisCommandTimeoutException("error in Redis");
            // 如果删除成功，则确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            int retryCount= (int) message.getMessageProperties().getHeaders().getOrDefault("x-retry-count",0);
            log.info("retryCount:"+retryCount);
            if(retryCount<MAX_RETRY_COUNT){
                retryCount++;
                // 更新重试次数并重新投递消息
                sendCacheDeletionMessage(pattern,retryCount);
                try {
                    // 确认旧消息已被处理，将其从队列中移除
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e1) {
                    log.error("Failed to acknowledge message after retry.", e1);
                }
            }else {
                log.error("Exceeded max retry count,"+pattern);
                try {
                    // 处理失败，拒绝消息且不重新投递
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                } catch (IOException e2) {
                    log.error("Failed to nack message after max retries.", e);
                }
            }
        }
    }

    private void sendCacheDeletionMessage(String pattern,int retryCount) {
        rabbitTemplate.convertAndSend(CACHE_TO_DEL_EXCHANGE,"cacheDel", pattern, msg -> {
            msg.getMessageProperties().setHeader("x-retry-count", retryCount);
            return msg;
        });
    }
}
